<?php
/**
 * @copyright  Copyright  LexisNexis, a division of Reed Elsevier Inc. All rights reserved.
 */
namespace App\Console\Commands;

use Illuminate\Console\Command;
use Illuminate\Support\Facades\Mail;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
use Elasticsearch\ClientBuilder;
use Elasticsearch\Client;

class QueueCommand extends Command {
    /**
     * 控制台命令名称
     *
     * @var string
     */
    protected $signature = 'queue:mq-to-es';

    /**
     * 控制台命令描述
     *
     * @var string
     */
    protected $description = 'Get data from MQ and push it to ES';

    private $exchangeName;
    private $queueName;
    private $connection;
    private $channel;
    /**
     * @var Client
     */
    private $esClient;


    public function __construct() {
        parent::__construct();
    }

    private function format_data($data) {
        /*
        if (!empty($body['vars']['session'])) {
            $this->stringArray($body['vars']['session']);
        }
        if (!empty($body['vars']['get'])) {
            $this->stringArray($body['vars']['get']);
        }
        if (!empty($body['vars']['post'])) {
            $this->stringArray($body['vars']['post']);
        }
        */
        /**
        [12] => products
        [14] => tabs
        [15] => orders
        [18] => pageLast
        [21] => articles
        [22] => suggestion
        [23] => boxes
        [24] => filter
         */

        $head = array();
        $body = array();
        $heads = array(
            "uid", "user_agent", "client_ip", "server_ip", "session_id", "real_session", "session_time",
            "prid", "crid", "url", "class", "create_time", "timestamp", "is_ajax", "is_js_event", "environment",
            "browser_name", "browser_version", "platform", "is_mobile", "is_robot", "is_tablet",
            "product", "name", "type",
        );
        $bodys = array(
            "url", "language", "mouseX", "mouseY", "clientW", "clientH", "scrollTop", "documentW", "documentH",
            "request_time", "tab", "page", "keyword", "num", "size", "order", "abstract", "current",
            "clicked_name", "clicked_url", "menu", "content_type"
        );
        if (!empty($data['head'])) {
            foreach ($data['head'] as $key => $value) {
                if (in_array($key, $heads)) {
                    $head[$key] = $value;
                    unset($data['head'][$key]);
                }
            }
            //$head['extra_head'] = json_encode($data['head']);
        }

        if (!empty($data['body'])) {
            $this->stringArray($data['body']);
            foreach ($data['body'] as $key => $value) {
                if (in_array($key, $bodys)) {
                    $body[$key] = $value;
                    unset($data['body'][$key]);
                }
            }
            $body['extra'] = json_encode($data['body']);
        }
        $result = array_merge($head, $body);
        //var_dump($result);
        return $result;
    }

    private function process_message(AMQPMessage $message) {
        if (!$message) {
            return false;
        }
        $bodyStr = $message->body;

        $body = json_decode($bodyStr, true);
        if (empty($body['id'])) {
            $this->info($bodyStr);
            $this->alert('bad body data.');
            return false;
        }
        $id = $body['id'];

        $body = $this->format_data($body);

        try {
            $month = '_' . date('Y_m');
            $indexName = env('ES_INDEX') . $month;
            $typeName = env('ES_TYPE') . $month;

            $exist = $this->esClient->indices()->exists([
                'index' => $indexName,
            ]);
            if (!$exist) {
                $result = $this->createIndex($indexName, $typeName);
                if (!$result) {
                    $this->error("create es index failed! index: $indexName");
                    $this->sendAlertEmail('PSS ALERT', "create es index failed! index: $indexName");
                    return false;
                }
            }

            $response = $this->esClient->index([
                'index' => $indexName,
                'type'  => $typeName,
                'id'    => $id,
                'body'  => $body
            ]);
        } catch (\Exception $exception) {
            $this->info($bodyStr);
            $this->error('index exception: ' . $exception->getMessage());
            $this->sendAlertEmail('PSS ALERT',
                'index exception: ' . $exception->getMessage() . "\n$bodyStr\n"
            );
            return false;
        }
        if (empty($response['created'])) {
            $this->info($bodyStr);
            $this->error('index failed!');
            $this->sendAlertEmail('PSS ALERT', "index failed!\n$bodyStr\n");
            return false;
        }

        return true;
    }

    private function stringArray(&$arr) {
        if (!is_array ($arr)) {
            return false;
        }

        foreach ($arr as $key => &$val) {
            if (is_array ($val)) {
                $this->stringArray($val);
            } else {
                $val = (string) $val;
                if (stripos($key, 'time') !== false && empty($val)) {
                    $val = '1970-01-01 00:00:00';
                } else if (stripos($key, 'date') !== false && empty($val)) {
                    $val = '1970-01-01';
                }
            }
        }
    }

    private function createIndex($indexName, $typeName) {
        $params = [
            'index' => $indexName,
            'body' => [
                'settings' => [
                    //每个索引的主分片数，默认值是 5
                    'number_of_shards' => 5,
                    //每个主分片的副本数，默认值是 1
                    'number_of_replicas' => 1,
                    //mapping总字段数
                    "index.mapping.total_fields.limit" => 10000
                ],
                "mappings" => [
                    $typeName => [
                        "properties" => [
                            "uid" => [
                                "type" => "integer"
                            ],
                            "user_agent" => [
                                "type" => "keyword"
                            ],
                            "client_ip" => [
                                "type" => "keyword"
                            ],
                            "server_ip" => [
                                "type" => "keyword"
                            ],
                            "session_id" => [
                                "type" => "keyword"
                            ],
                            "real_session" => [
                                "type" => "keyword"
                            ],
                            "session_time" => [
                                "type" => "integer"
                            ],
                            "prid" => [
                                "type" => "keyword"
                            ],
                            "crid" => [
                                "type" => "keyword"
                            ],
                            "url" => [
                                "type" => "keyword"
                            ],
                            "class" => [
                                "type" => "keyword"
                            ],
                            "create_time" => [
                                "type" => "date",
                                "format" => "yyyy-MM-dd HH:mm:ss"
                            ],
                            "timestamp" => [
                                "type" => "integer"
                            ],
                            "is_js_event" => [
                                "type" => "integer"
                            ],
                            "environment" => [
                                "type" => "keyword"
                            ],
                            "browser_name" => [
                                "type" => "keyword"
                            ],
                            "browser_version" => [
                                "type" => "keyword"
                            ],
                            "platform" => [
                                "type" => "keyword"
                            ],
                            "is_mobile" => [
                                "type" => "integer"
                            ],
                            "is_robot" => [
                                "type" => "integer"
                            ],
                            "is_tablet" => [
                                "type" => "integer"
                            ],
                            "product" => [
                                "type" => "keyword"
                            ],
                            "name" => [
                                "type" => "keyword"
                            ],
                            "type" => [
                                "type" => "keyword"
                            ],
                            "language" => [
                                "type" => "keyword"
                            ],
                            "mouseX" => [
                                "type" => "integer"
                            ],
                            "mouseY" => [
                                "type" => "integer"
                            ],
                            "clientW" => [
                                "type" => "integer"
                            ],
                            "clientH" => [
                                "type" => "integer"
                            ],
                            "scrollTop" => [
                                "type" => "integer"
                            ],
                            "documentW" => [
                                "type" => "integer"
                            ],
                            "documentH" => [
                                "type" => "integer"
                            ],
                            "extra" => [
                                "type" => "text"
                            ],
                            "request_time" => [
                                "type" => "keyword"
                            ],
                            "tab" => [
                                "type" => "keyword"
                            ],
                            "page" => [
                                "type" => "keyword"
                            ],
                            "keyword" => [
                                "type" => "keyword"
                            ],
                            "num" => [
                                "type" => "keyword"
                            ],
                            "size" => [
                                "type" => "keyword"
                            ],
                            "order" => [
                                "type" => "keyword"
                            ],
                            "abstract" => [
                                "type" => "keyword"
                            ],
                            "current" => [
                                "type" => "keyword"
                            ],
                            "clicked_name" => [
                                "type" => "keyword"
                            ],
                            "clicked_url" => [
                                "type" => "keyword"
                            ],
                            "menu" => [
                                "type" => "keyword"
                            ],
                            "content_type" => [
                                "type" => "keyword"
                            ]
                        ]
                    ]
                ]
            ]
        ];

        //[acknowledged] => 1
        $response = $this->esClient->indices()->create($params);
        if (!isset($response['acknowledged']) || $response['acknowledged'] != 1) {
            return false;
        }

        return true;
    }

    /**
     * 执行控制台命令
     *
     * @return mixed
     */
    public function handle() {
        $this->info('begin.');

        $esHost = [
            'host' => env('ES_HOST'),
            'port' => env('ES_PORT'),
            'user' => env('ES_USER'),
            'pass' => env('ES_PASS')
        ];
        $this->esClient = ClientBuilder::create()->setHosts([$esHost])->build();

        if (env('AMQP_DEBUG') == 'true') {
            define('AMQP_DEBUG', true);
        }
        $this->exchangeName = env('MQ_EXCHANGE_NAME');
        $this->queueName = env('MQ_QUEUE_NAME');

        $this->connection = new AMQPStreamConnection(
            env('MQ_HOST'), env('MQ_PORT'), env('MQ_USER'), env('MQ_PASS'), env('MQ_VHOST')
        );
        $this->channel = $this->connection->channel();
        //passive:是否被动，主动会自动建队列 durable:持久化 exclusive:是否私有 auto_delete:取消订阅自动删除队列
        $this->channel->queue_declare($this->queueName, false, true, false, false);
        $this->channel->exchange_declare($this->exchangeName, 'direct', false, true, false);
        $this->channel->queue_bind($this->queueName, $this->exchangeName);

        $prefetchSize = null; // msg size in bytes, must be null else error
        $prefetchCount = 1;
        $applyPerChannel = null; // can be false or null, otherwise error
        $this->channel->basic_qos($prefetchSize, $prefetchCount, $applyPerChannel);

        $consumerTag = 'consumer';
        $this->channel->basic_consume(
            $this->queueName,
            $consumerTag,
            false, false, false, false,
            function (AMQPMessage $message) {
                $this->process_message($message);
                $message->delivery_info['channel']->basic_ack($message->delivery_info['delivery_tag']);
            }
        );

        //count($this->channel->callbacks)
        for ($i = 0; $i <= 1000; $i++) {
            try {
                $this->channel->wait(null, false, 3);
            } catch (\Exception $exception) {
                sleep(1);//@todo should to be 60 in PRD
                break;
            }
        }

        $this->channel->close();
        $this->connection->close();

        $this->info('done.');
    }

    protected function sendAlertEmail($subject, $content) {
        $toMail  = explode(';', env('MAIL_TO_LIST'));
        $fromMail = env('MAIL_FROM_ADDRESS');

        Mail::raw($content, function ($message) use ($subject, $toMail, $fromMail) {
            $message->subject($subject);
            $message->to($toMail);
            $message->from($fromMail);
        });
    }
}